﻿using CanalService.Entities;
using CanalService.Model;
using CanalSharp.Connections;
using CanalSharp.Protocol;
using Google.Protobuf.Collections;
using MongoDB.Driver;
using System.Reflection;

namespace CanalService.Services
{
    public class CanalToMySqlService
    {
        private readonly IUserInfoRepository _userRepository;

        public CanalToMySqlService(IUserInfoRepository repository)
        {
            _userRepository = repository;
        }



        /// <summary>
        /// 启动同步
        /// </summary>
        public async void OnStart()
        {

            //1、开启日志：
            var loggerFactory = LoggerFactory.Create(builder =>
            {
                builder.AddFilter("Microsoft", LogLevel.Debug)
               .AddFilter("System", LogLevel.Information)
               .AddConsole();
            });


            //1.1 监听Canal的调用日志
            var logger = loggerFactory.CreateLogger<SimpleCanalConnection>();


            //2、建立Canal服务器连接，默认端口是11111，
            var canalConn = new SimpleCanalConnection(new SimpleCanalOptions("127.0.0.1", 11111, "101"), logger);
            await canalConn.ConnectAsync();

            //3、订阅数据库和表
            await canalConn.SubscribeAsync("ndc_test.canal_table");

            while (true)
            {
                try
                {
                    //4.1 分段读取Canal数据，fetchSize 定义了一次获取的数据量
                    var canalData = await canalConn.GetWithoutAckAsync(1024);

                    //4.2 判断数据类型(加锁或不加锁)
                    var entries = canalData.Entries;
                    foreach (var entry in entries)
                    {
                        //不处理事务标记
                        if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
                        {
                            continue;
                        }
                        //4.3 判断数据操作类型：增删改查
                        RowChange rowChange = RowChange.Parser.ParseFrom(entry.StoreValue);
                        switch (rowChange.EventType)
                        {
                            case EventType.Compatibleproto2:
                                break;
                            case EventType.Insert:
                                logger.LogInformation("============增加数据============");
                                await HandleCanalInsertDataAsync(rowChange.RowDatas, entry);
                                break;
                            case EventType.Update:
                                logger.LogInformation("============修改数据============");
                                break;
                            case EventType.Delete:
                                logger.LogInformation("============删除数据============");
                                break;
                            case EventType.Create:
                                logger.LogInformation("============创建表============");
                                break;
                            case EventType.Alter:
                                logger.LogInformation("============修改表============");
                                break;
                            case EventType.Erase:
                                logger.LogInformation("============Erase表============");
                                break;
                            case EventType.Query:
                                logger.LogInformation("===========查询数据============="); 
                                break;
                            case EventType.Truncate:
                                break;
                            case EventType.Rename:
                                break;
                            case EventType.Cindex:
                                break;
                            case EventType.Dindex:
                                break;
                            case EventType.Gtid:
                                break;
                            case EventType.Xacommit:
                                break;
                            case EventType.Xarollback:
                                break;
                            case EventType.Mheartbeat:
                                break;
                            default:
                                break;
                        }
                    }


                    // 确认已处理的数据，避免重复处理
                    await canalConn.AckAsync(canalData.Id);

                }
                catch (Exception e)
                {
                    logger.LogError($"同步数据异常：{e.Message} Exception:", e);
                }
            }
        }
 

        /// <summary>
        /// 处理Canal新增数据变动
        /// </summary>
        /// <returns></returns>
        private async Task HandleCanalInsertDataAsync(RepeatedField<RowData> rowDatas, Entry entry)
        {
            foreach (RowData rowData in rowDatas)
            {
                //1.获取需要同步的业务数据
                List<Column> columns = rowData.AfterColumns.ToList();
                //2.获取数据库
                var schemaName = entry.Header.SchemaName;
                //3.获取表
                var tableName = entry.Header.TableName;
                //4.根据表名走不同业务
                if (tableName == "usertest")
                {
                    await Console.Out.WriteLineAsync("同步usertest表。。。");
                }
                if (tableName == "canal_table")
                {
                    await Console.Out.WriteLineAsync("同步canal_table表");

                    InsertUserTestToMySql(columns);
                }
            }
        }


        /// <summary>
        /// 同步数据到Mongo
        /// </summary>
        /// <param name="columns"></param>
        private void InsertUserTestToMySql(List<Column> columns)
        {
            try
            {
                CanalTable model = new();
                // 1、数据转换
                // Type
                System.Type type = typeof(CanalTable);
                // 2、取出具体数据
                foreach (var column in columns)
                {
                    // 3、取出字段名
                    string FiledName = column.Name;
                    // 4、取出字段值
                    string FiledValue = column.Value;
                    Console.WriteLine(FiledName + ":" + FiledValue);
                    // 5、属性赋值
                    PropertyInfo propertyInfo = type.GetProperty(FiledName);
                    // 6、数据类型转换
                    if (propertyInfo != null)
                    {
                        System.Type type1 = propertyInfo.PropertyType;
                        if (type1 == typeof(int))
                        {
                            propertyInfo.SetValue(model, int.Parse(FiledValue));
                        }
                        else if (type1 == typeof(string))
                        {
                            propertyInfo.SetValue(model, FiledValue);
                        }
                        else if (type1 == typeof(double))
                        {
                            propertyInfo.SetValue(model, double.Parse(FiledValue));
                        }
                        else if (type1 == typeof(bool))
                        {
                            propertyInfo.SetValue(model, bool.Parse(FiledValue));
                        }
                        else if (type1 == typeof(DateTime))
                        {
                            propertyInfo.SetValue(model, DateTime.Parse(FiledValue));
                        }
                        else if (type1 == typeof(int))
                        {

                            propertyInfo.SetValue(model, Int64.Parse(FiledValue));
                        }
                        else
                        {
                            Console.WriteLine("类型错误");
                        }
                    }
                }

                // 写入目标数据库

                //1、打开数据库连接 
                var userInfo = new UserInfo
                {
                    Id = model.Id,
                    UserName = model.Name,
                    CreateTime = model.CreatedDate
                };
                var insertResult =  _userRepository.InsertUserInfoAsync(userInfo).GetAwaiter().GetResult();
                if (insertResult)
                {
                    Console.WriteLine("------------------------------同步数据写入成功：================");
                }
                else
                {

                    Console.WriteLine("================同步数据写入失败：================");
                }




            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }
        }

    }
}
